package com.examples.mysql2mysql.batch;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class JdbcSinkFunction extends RichSinkFunction<Tuple2<String, Integer>> {
    private Connection connection;
    private PreparedStatement preparedStatement;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        super.open(parameters);

        // 创建数据库连接
        connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/f_saas_cfg?useUnicode=true&characterEncoding=UTF-8&useAffectedRows=true&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8", "root", "123456");

        // 准备插入语句
        preparedStatement = connection.prepareStatement("INSERT INTO `test_table01` (`name`, del) VALUES (?, ?)");
    }

    @Override
    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
        // 设置参数并执行插入
        preparedStatement.setString(1, value.f0);
        preparedStatement.setInt(2, value.f1);
        preparedStatement.execute();
    }

    @Override
    public void close() throws Exception {
        super.close();

        // 关闭连接和预处理语句
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
